/*
 * Copyright (C) 2017 Pingan, Inc. All Rights Reserved.
 */

package com.pingan.haofang.searchcloud.batch.export;

import com.alibaba.druid.pool.DruidDataSource;
import com.pingan.haofang.framework.common.functional.lang.control.base.Enums;
import com.pingan.haofang.searchcloud.api.matedata.constant.MetadataFieldType;
import com.pingan.haofang.searchcloud.api.route.AliasUtils;
import com.pingan.haofang.searchcloud.batch.common.Batch;
import com.pingan.haofang.searchcloud.batch.common.BatchResult;
import com.pingan.haofang.searchcloud.batch.common.BatchRunner;
import com.pingan.haofang.searchcloud.index.dto.IndexInfoQueryResult;
import com.pingan.haofang.searchcloud.index.service.IndexService;
import com.pingan.haofang.searchcloud.metadata.dto.MetadataFieldQueryResult;
import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.SolrDocument;
import org.apache.solr.common.SolrDocumentList;
import org.apache.solr.common.params.CursorMarkParams;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.jdbc.core.JdbcTemplate;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;

/**
 * Solr数据导入到MySQL,用于数据测试
 *
 * @author zhangbi617
 * @date 2017-07-10
 */
@Batch(value = "solrData2MySql")
@ConditionalOnProperty("solrData2MySql")
public class SolrData2MySqlRunner implements BatchRunner, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(SolrData2MySqlRunner.class);

    @Autowired
    CloudSolrClient solrClient;

    @Autowired
    IndexService indexService;

    private JdbcTemplate exportJdbcTemplate;

    private static final int ROWS = 1000;

    /**
     * 指定数据库所在机器IP端口
     * -Dsolr.export.mysql.host=xxx
     */
    @Value("${solr.export.mysql.host:10.59.78.21}")
    private String exportMysqlHost;
    /**
     * 指定MySQL用户名
     * -Dsolr.export.mysql.username=xxx
     */
    @Value("${solr.export.mysql.username:root}")
    private String exportMysqlUserName;
    /**
     * 指定MySQL密码
     * -Dsolr.export.mysql.password=xxx
     */
    @Value("${solr.export.mysql.password:123456}")
    private String exportMysqlPassword;

    /**
     * 可以指定要导出的indexName,多个逗号分隔
     * -Dsolr.export.indexNames=index1,index2
     */
    @Value("${solr.export.indexNames:ALL_INDEX}")
    private String exportIndexName;

    @Override
    public BatchResult run(String[] args) {
        List<String> aliases = getAliases();
        LOGGER.info("all alias:{}", aliases);

        for (String alias : aliases) {

            // 1. schema
            IndexInfoQueryResult schema = initIndexSchema(alias);

            // 2. data
            try {
                exportDocuments2DB(alias, schema);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }

            LOGGER.info("index for {} export success \n", alias);
        }
        return BatchResult.ALL_SUCCESS;
    }

    /**
     * 游标方式导出数据
     *
     * @param alias  别名
     * @param schema schema
     * @throws SolrServerException
     * @throws IOException
     */
    private void exportDocuments2DB(String alias, IndexInfoQueryResult schema) throws SolrServerException, IOException {

        String cursorMark = CursorMarkParams.CURSOR_MARK_START;

        boolean done = false;
        int count = 0;
        while (!done) {
            SolrQuery params = new SolrQuery("*:*");
            params.setRows(ROWS);
            params.set(CursorMarkParams.CURSOR_MARK_PARAM, cursorMark);
            params.setSort(SolrQuery.SortClause.asc(getUniqueKey(schema.getFields())));

            QueryResponse response = solrClient.query(alias, params);

            SolrDocumentList list = response.getResults();

            String nextCursorMark = response.getNextCursorMark();
            LOGGER.info("for {}, numFound:{}, return:{}, nextCursor:{}, count:{}",
                    alias, list.getNumFound(), list.size(), nextCursorMark, count);

            saveDocumentList2DB(list, schema);

            if (cursorMark.equals(nextCursorMark)) {
                done = true;
            }
            cursorMark = nextCursorMark;
            count += list.size();
        }

    }

    /**
     * 数据存到数据库
     *
     * @param list
     * @param schema
     */
    private void saveDocumentList2DB(SolrDocumentList list, IndexInfoQueryResult schema) {
        StringBuilder insertSB = new StringBuilder();
        insertSB.append("insert into `" + schema.getBasicInfo().getIndexName() + "` (");
        for (int i = 0; i < schema.getFields().size(); i++) {
            MetadataFieldQueryResult field = schema.getFields().get(i);
            insertSB.append("`" + field.getFieldName() + "`");
            if (i < schema.getFields().size() - 1) {
                insertSB.append(",");
            }
        }
        insertSB.append(") values (" + generateParamHolder(schema.getFields().size()) + ")");

        LOGGER.info("insert sql:{}", insertSB.toString());

        List<Object[]> params = new ArrayList<>();
        ListIterator it = list.listIterator();
        while (it.hasNext()) {
            SolrDocument document = (SolrDocument) it.next();
            Object[] param = new Object[schema.getFields().size()];
            for (int i = 0; i < schema.getFields().size(); i++) {
                MetadataFieldQueryResult field = schema.getFields().get(i);
                Object object = document.getFieldValue(field.getFieldName());
                if (object == null) {
                    param[i] = null;
                    continue;
                }
                if (field.isMultiValued()) {
                    param[i] = fieldValueToString(object);
                } else {
                    MetadataFieldType metadataFieldType = Enums.findValue(field.getFieldTypeId(),
                            MetadataFieldType.class);
                    switch (metadataFieldType) {
                        case INT:
                        case LONG:
                            param[i] = Integer.valueOf(object.toString());
                        case FLOAT:
                        case DOUBLE:
                            param[i] = Double.valueOf(object.toString());
                            break;
                        case GEO:
                        case STRING:
                        case MY_IK:
                        default:
                            param[i] = fieldValueToString(object);
                            break;
                    }
                }
            }
            params.add(param);
        }
        exportJdbcTemplate.batchUpdate(insertSB.toString(), params);
    }

    private String fieldValueToString(Object o) {
        try {
            return StringUtils.substring(o.toString(), 0, 1000);
        } catch (Exception e) {
//            LOGGER.warn(e.getMessage(), e);
            return null;
        }
    }

    private String getUniqueKey(List<MetadataFieldQueryResult> fields) {
        for (MetadataFieldQueryResult field : fields) {
            if (field.isUniqueKey()) {
                return field.getFieldName();
            }
        }
        return "id";
    }

    private String generateParamHolder(int size) {
        List<String> holderList = new ArrayList<>();
        for (int i = 0; i < size; i++) {
            holderList.add("?");
        }
        return StringUtils.join(holderList, ",");
    }

    /**
     * 初始化DB
     *
     * @param alias
     */
    private IndexInfoQueryResult initIndexSchema(String alias) {
        String indexName = parseIndexNameFromAlias(alias);
        IndexInfoQueryResult result = indexService.queryByIndexNameOfStorage(indexName);
        String dropSchema = String.format("drop table if exists `%s`;", indexName);
        LOGGER.info("drop schema:{}", dropSchema);
        exportJdbcTemplate.execute(dropSchema);

        // 拼接create table sql
        StringBuilder createSb = new StringBuilder("create table `" + indexName + "` (");
        for (int i = 0; i < result.getFields().size(); i++) {
            MetadataFieldQueryResult field = result.getFields().get(i);
            createSb.append("`" + field.getFieldName() + "` " + getType(field.getFieldTypeId(), field.isMultiValued(), field.isUniqueKey()) + getNULL(field));
            createSb.append(",");
        }
        createSb.append("PRIMARY KEY (`" + getUniqueKey(result.getFields()) + "`)");
        createSb.append(" ) ");
        createSb.append("ENGINE=MyISAM COMMENT = '" + alias + "'");

        LOGGER.info("create schema: {}", createSb.toString());
        exportJdbcTemplate.execute(createSb.toString());
        return result;
    }

    private String getNULL(MetadataFieldQueryResult field) {
        if (field.isRequired() || field.isUniqueKey()) {
            return " NOT NULL ";
        } else {
            return " NULL ";
        }
    }

    /**
     * sql field type
     *
     * @param fieldTypeId
     * @param multiValued
     * @param uniqueKey
     * @return
     */
    private String getType(int fieldTypeId, boolean multiValued, boolean uniqueKey) {
        if (multiValued) {
            return " VARCHAR(1000) ";
        }
        MetadataFieldType metadataFieldType = Enums.findValue(fieldTypeId, MetadataFieldType.class);
        switch (metadataFieldType) {
            case INT:
                return " INT(11) ";
            case LONG:
                return " BIGINT(20) ";
            case FLOAT:
                return " FLOAT ";
            case DOUBLE:
                return " DOUBLE ";
            case GEO:
            case STRING:
            case MY_IK:
            default:
                if (uniqueKey) {
                    return " VARCHAR(100) ";
                }
                return " VARCHAR(1000) ";
        }
    }


    /**
     * 别名转为索引名
     *
     * @param alias
     * @return
     */
    private String parseIndexNameFromAlias(String alias) {
        return alias.replaceFirst("searchcloud_", "");
    }


    /**
     * 获取所有别名
     *
     * @return
     */
    private List<String> getAliases() {
        // 可以指定具体的index
        if (!"ALL_INDEX".equals(this.exportIndexName)) {
            String[] indexNames = this.exportIndexName.split(",");
            List<String> aliasSet = new ArrayList<>();
            for (String indexName : indexNames) {
                aliasSet.add(AliasUtils.convertAliasName(indexName));
                return aliasSet;
            }
        }
        if (solrClient.getZkStateReader() == null) {
            solrClient.connect();
        }

        // 以下拿到alias，按照数据量从小到大排序
        Map<String, String> aliasKVMap = solrClient.getZkStateReader().getAliases()
                .getCollectionAliasMap();
        List<AliasBo> aliasBos = new ArrayList<>();
        for (Map.Entry<String, String> entry : aliasKVMap.entrySet()) {
            AliasBo aliasBo = new AliasBo();
            aliasBo.alias = entry.getKey();
            aliasBo.collection = entry.getValue();
            SolrQuery params = new SolrQuery("*:*");
            params.setRows(1);
            params.setStart(1);

            QueryResponse response = null;
            try {
                response = solrClient.query(entry.getKey(), params);
                aliasBo.size = (int) response.getResults().getNumFound();
            } catch (Exception e) {
                LOGGER.info(e.getMessage(), e);
            }
            aliasBos.add(aliasBo);
        }

        Collections.sort(aliasBos);


        List<String> list = new ArrayList<>();
        for (AliasBo aliasBo : aliasBos) {
            list.add(aliasBo.alias);
        }
        LOGGER.info("alias : {}", list);
        return list;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        DruidDataSource druidDataSource = new DruidDataSource();
        druidDataSource.setUrl("jdbc:mysql://" + exportMysqlHost + "/searchcloud_index_db?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&rewriteBatchedStatements=false");
        druidDataSource.setUsername(exportMysqlUserName);
        druidDataSource.setPassword(exportMysqlPassword);
        druidDataSource.init();
        this.exportJdbcTemplate = new JdbcTemplate(druidDataSource);
    }

    private static class AliasBo implements Comparable<AliasBo> {
        private String alias;
        private String collection;
        private int size = 0;

        @Override
        public int compareTo(AliasBo o) {
            return this.size - o.size;
        }

        @Override
        public String toString() {
            return "AliasBo{" +
                    "alias='" + alias + '\'' +
                    ", collection='" + collection + '\'' +
                    ", size=" + size +
                    '}';
        }
    }
}
